feat: add public queue admin APIs (requeue, shutdown, resources)#405
Merged
Conversation
Atomically resets a stuck Running job to Pending, releases its execution claim, and clears any stale cancel request, without consuming retry budget. No automatic path recovers a Running job whose worker is hung-but-heartbeating or whose claim row is gone.
Also forwards the default through delay() and map(), which never passed expires before.
📝 WalkthroughWalkthroughChangesThe storage layer now supports atomically requeueing stuck running jobs. Python exposes job requeueing, task expiry defaults, resource reload and argument analysis, predicate statistics, and graceful shutdown APIs, with synchronous/asynchronous wrappers, tests, type stubs, and documentation updates. Stuck Job Recovery
Python Queue API Expansion
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant Queue
participant Storage
participant Backend
Queue->>Storage: requeue_job(job_id)
Storage->>Backend: requeue_stuck(job_id, now)
Backend->>Backend: move Running job to Pending
Backend->>Backend: clear execution claim
Backend-->>Storage: return boolean
Storage-->>Queue: return result
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
sdks/python/taskito/mixins/decorators.py (1)
360-366: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winMove
expiresto the end or make the decorator keyword-only. Inserting it beforeprioritychanges every existing positional@queue.task(...)call: later arguments likepriority,rate_limit, andqueuewill shift intoexpires, causing silent behavior changes.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@sdks/python/taskito/mixins/decorators.py` around lines 360 - 366, Update the task decorator signature in task so the newly added expires parameter is placed after all existing positional parameters, or make the decorator parameters keyword-only; preserve the current positional ordering for priority, rate_limit, queue, and other existing arguments.
🧹 Nitpick comments (2)
sdks/python/taskito/task.py (1)
23-34: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueSame mid-signature insertion pattern as
QueueDecoratorMixin.task().
default_expiresis inserted beforeinjectin a positional-or-keyword__init__. Lower risk since the only known constructor call site (decorators.py) uses kwargs, butTaskWrapperis a public type — worth applying the same fix (append at end or keyword-only) for consistency.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@sdks/python/taskito/task.py` around lines 23 - 34, Update TaskWrapper.__init__ so default_expires does not precede the existing inject parameter in the positional-or-keyword signature; move default_expires after inject or make it keyword-only, and update any affected calls while preserving backward compatibility.sdks/python/tests/core/test_expires.py (1)
1-88: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueSolid coverage of the default/override/fallback logic and an end-to-end expiry check. One gap: no test exercises
.map()forwardingexpires(onlydelay/apply_asyncare covered), though it shares the same_default_expiresfield so risk is low.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@sdks/python/tests/core/test_expires.py` around lines 1 - 88, Add a test covering task-level expiry propagation through the task’s `.map()` API, using a task configured with an `expires` default, invoking `map()`, and asserting the capture middleware receives that expiry value for the enqueued jobs. Place it alongside the existing expiry forwarding tests and use `_queue_with_capture` and `_CaptureEnqueue`.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@sdks/python/taskito/mixins/decorators.py`:
- Around line 360-366: Update the task decorator signature in task so the newly
added expires parameter is placed after all existing positional parameters, or
make the decorator parameters keyword-only; preserve the current positional
ordering for priority, rate_limit, queue, and other existing arguments.
---
Nitpick comments:
In `@sdks/python/taskito/task.py`:
- Around line 23-34: Update TaskWrapper.__init__ so default_expires does not
precede the existing inject parameter in the positional-or-keyword signature;
move default_expires after inject or make it keyword-only, and update any
affected calls while preserving backward compatibility.
In `@sdks/python/tests/core/test_expires.py`:
- Around line 1-88: Add a test covering task-level expiry propagation through
the task’s `.map()` API, using a task configured with an `expires` default,
invoking `map()`, and asserting the capture middleware receives that expiry
value for the enqueued jobs. Place it alongside the existing expiry forwarding
tests and use `_queue_with_capture` and `_CaptureEnqueue`.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: c4cefe76-6026-4071-b4a8-6c1b58c8abb5
📒 Files selected for processing (31)
crates/taskito-core/src/storage/diesel_common/jobs.rscrates/taskito-core/src/storage/mod.rscrates/taskito-core/src/storage/redis_backend/jobs/state.rscrates/taskito-core/src/storage/traits.rscrates/taskito-core/tests/rust/storage_tests.rscrates/taskito-python/src/py_queue/inspection.rsdocs/content/docs/python/api-reference/queue/index.mdxdocs/content/docs/python/api-reference/queue/queues.mdxdocs/content/docs/python/api-reference/queue/resources.mdxdocs/content/docs/python/api-reference/queue/workers.mdxdocs/content/docs/python/guides/resources/configuration.mdxdocs/content/docs/shared/guides/core/predicates.mdxdocs/content/docs/shared/guides/core/tasks.mdxdocs/content/docs/shared/guides/core/workers.mdxdocs/content/docs/shared/guides/operations/troubleshooting.mdxdocs/content/docs/shared/guides/resources/interception.mdxsdks/python/taskito/_taskito.pyisdks/python/taskito/async_support/mixins.pysdks/python/taskito/mixins/decorators.pysdks/python/taskito/mixins/lifecycle.pysdks/python/taskito/mixins/operations.pysdks/python/taskito/mixins/predicates.pysdks/python/taskito/mixins/resources.pysdks/python/taskito/task.pysdks/python/tests/conftest.pysdks/python/tests/core/test_expires.pysdks/python/tests/core/test_predicates_enqueue.pysdks/python/tests/core/test_requeue.pysdks/python/tests/core/test_shutdown.pysdks/python/tests/resources/test_interception.pysdks/python/tests/resources/test_resources.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
requeue_job,shutdown,reload_resources,predicate_stats,analyze_arguments) replacing the internal underscore-prefixed accessors the docs had flagged as internalrequeue_stuckcore storage op to recover Running jobs that no automatic mechanism touches (hung-but-heartbeating worker, huge timeout, or a claim-less Running job)queue.task(expires=)as a per-task default;delay()/map()now forward expires (previously never passed)Changes
requeue_stuck(id, now)in diesel_common (SQLite+Postgres, single transaction) and a Redis Lua script (SISMEMBER Running-set guard). Atomically resets a stuck Running job to Pending, deletes its execution claim, clears a stale cancel request, preserves retry budget. Contract testtest_requeue_stuckruns against all 3 backends.requeue_job/arequeue_job,shutdown()(programmatic graceful shutdown),reload_resources/areload_resources(programmatic hot reload;{}when no runtime active),predicate_stats()(allowed/denied/deferred/cancelled/errors),analyze_arguments() -> InterceptionReport(queue-level interception dry-run; empty when interception off).expiresfollows thedefault_timeoutplumbing.Test plan
Summary by CodeRabbit
New Features
queue.requeue_job()to recover stuck running jobs, release stale claims, preserve retry budgets, and prevent duplicate cancellation.queue.shutdown(), resource hot reload, argument analysis, predicate statistics, and async counterparts.Documentation
Tests